import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.HamaConfiguration;
import org.apache.hama.bsp.BSP;
import org.apache.hama.bsp.BSPJob;
import org.apache.hama.bsp.BSPJobClient;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.ClusterStatus;
import org.apache.hama.bsp.FileOutputFormat;
import org.apache.hama.bsp.NullInputFormat;
import org.apache.hama.bsp.TextOutputFormat;
import org.apache.hama.bsp.sync.SyncException;

public class PiEstimator {
	private static Path TMP_OUTPUT = new Path("/tmp/pi-" + System.currentTimeMillis());

	public static class MyEstimator extends BSP<NullWritable, NullWritable, Text, DoubleWritable, DoubleWritable> {
		public static final Log LOG = LogFactory.getLog(MyEstimator.class);
		private String masterTask;
		private static final int iterations = 100000;

		@Override
		public void bsp(final BSPPeer<NullWritable, NullWritable, Text, DoubleWritable, DoubleWritable> peer) throws IOException,
				SyncException, InterruptedException {

			MyEstimator.LOG.info("peers: " + peer.getNumPeers());

			int in = 0;
			for (int i = 0; i < MyEstimator.iterations; i++) {
				final double x = (2.0 * Math.random()) - 1.0, y = (2.0 * Math.random()) - 1.0;
				if (Math.sqrt((x * x) + (y * y)) < 1.0) {
					in++;
				}
			}

			final double data = (4.0 * in) / MyEstimator.iterations;

			peer.send(this.masterTask, new DoubleWritable(data));
			peer.sync();
		}

		@Override
		public void setup(final BSPPeer<NullWritable, NullWritable, Text, DoubleWritable, DoubleWritable> peer) throws IOException {
			// Choose one as a master
			this.masterTask = peer.getPeerName(peer.getNumPeers() / 2);
		}

		@Override
		public void cleanup(final BSPPeer<NullWritable, NullWritable, Text, DoubleWritable, DoubleWritable> peer) throws IOException {
			if (peer.getPeerName().equals(this.masterTask)) {
				double pi = 0.0;
				final int numPeers = peer.getNumCurrentMessages();
				DoubleWritable received;
				while ((received = peer.getCurrentMessage()) != null) {
					pi += received.get();
				}

				pi = pi / numPeers;
				peer.write(new Text("Estimated value of PI is"), new DoubleWritable(pi));
				peer.write(new Text("nb. peers: " + numPeers), new DoubleWritable(pi));
			}
		}
	}

	static void printOutput(final HamaConfiguration conf) throws IOException {
		final FileSystem fs = FileSystem.get(conf);
		final FileStatus[] files = fs.listStatus(PiEstimator.TMP_OUTPUT);
		for (final FileStatus file : files) {
			if (file.getLen() > 0) {
				final FSDataInputStream in = fs.open(file.getPath());
				IOUtils.copyBytes(in, System.out, conf, false);
				in.close();
				break;
			}
		}

		// fs.delete(TMP_OUTPUT, true);
	}

	public static void main(final String[] args) throws InterruptedException, IOException, ClassNotFoundException {
		// BSP job configuration
		final HamaConfiguration conf = new HamaConfiguration();

		final BSPJob bsp = new BSPJob(conf, PiEstimator.class);
		// Set the job name
		bsp.setJobName("Pi Estimation Example");
		bsp.setBspClass(MyEstimator.class);
		bsp.setInputFormat(NullInputFormat.class);
		bsp.setOutputKeyClass(Text.class);
		bsp.setOutputValueClass(DoubleWritable.class);
		bsp.setOutputFormat(TextOutputFormat.class);
		FileOutputFormat.setOutputPath(bsp, PiEstimator.TMP_OUTPUT);

		final BSPJobClient jobClient = new BSPJobClient(conf);
		final ClusterStatus cluster = jobClient.getClusterStatus(true);

		if (args.length > 0) {
			bsp.setNumBspTask(Integer.parseInt(args[0]));
		} else {
			// Set to maximum
			bsp.setNumBspTask(cluster.getMaxTasks());
		}

		final long startTime = System.currentTimeMillis();
		if (bsp.waitForCompletion(true)) {
			PiEstimator.printOutput(conf);
			System.out.println("Job Finished in " + ((System.currentTimeMillis() - startTime) / 1000.0) + " seconds");
		}
	}
}
